package com.tyc.kafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class CustomProducerCallbackPartition {
    public static void main(String[] args) throws InterruptedException {
        // 属性配置
        Properties properties = new Properties();
        // 连接地址 多个用逗号隔开
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092,127.0.0.1:9093");
        // 指定序列化类型
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG,"0");
        // 创建生产者
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        // 发送数据
        for (int i = 0; i < 500; i++) {
            /**
             * 指定key: 跟据hashcode值与分区数取余，得到分区
             * 指定:partition
             * 未指定分区也未指定key,往一个分区发数据直到batch(batch.size=16k 默认)放满或者查过设置的时间(linger.ms(5-100ms)默认0毫秒),
             * 进行计算发往下个分区,每次分区与上次不一样
             */
            kafkaProducer.send(new ProducerRecord<>("test_g","hello bc:" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if(null == e){
                        System.out.println(String.format("主题：%s,分区：%s",recordMetadata.topic(),recordMetadata.partition()));
                    }

                }
            });
            // 测试未指定分区也未指定key，超时后重新计算下一个分区
            TimeUnit.MILLISECONDS.sleep(1);
        }
        //关闭资源
        kafkaProducer.close();
    }
}
